


/*
 * test_pipe.cpp
 *
 *  Created on: Jun 26, 2014
 *      Author: killerg
 */

#include <event2/util.h>
#include <sys/epoll.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>

struct cmd_t {
	void* arg2;
	int arg1;
	char type;
};

struct thread_t {
	int wfd;
	int rfd;
};

void producer(int wfd) {
	cmd_t cmd;
	for (int i=0; i<1000000000; ++i) {
		cmd.type = 0;
		cmd.arg1 = i;
		cmd.arg2 = NULL;

		int size = sizeof(cmd_t);
		cmd_t cmd;
		int readed = 0;

		int count = 0;
		do {
			int len = write(wfd, &cmd, size - readed);
			if (len <= 0) {
				perror("read");
				exit(EXIT_FAILURE);
			}
			readed += len;
		} while (readed < size);
	}
}

void consumer(int rfd) {
	int readtimes[100] = {0};
	int totalcount = 0;

	int epollfd = epoll_create(10);
	if (epollfd == -1) {
		perror("epoll_create");
		exit(EXIT_FAILURE);
	}

	epoll_event ev;
	ev.events = EPOLLIN;
	ev.data.fd = rfd;
	if (epoll_ctl(epollfd, EPOLL_CTL_ADD, rfd, &ev) == -1) {
		perror("epoll_ctl: listen_sock");
		exit(EXIT_FAILURE);
	}

	epoll_event events[1];

	for (;;) {
		int nfds = epoll_wait(epollfd, events, 1, -1);
		if (nfds == -1) {
			perror("epoll_pwait");
			exit(EXIT_FAILURE);
		}

		if (events->events & EPOLLIN) {
			int size = sizeof(cmd_t);
			cmd_t cmd;
			uint8_t* ptr = (uint8_t*)&cmd;
			int readed = 0;

			int count = 0;
			do {
				int len = read(rfd, ptr+readed, size - readed);
				if (len <= 0) {
					perror("read");
					exit(EXIT_FAILURE);
				}
				readed += len;
				++count;
			} while (readed < size);
			readtimes[count]++;
			totalcount++;
		}
		if (totalcount % 1000000 == 0) {
			for (int i=0; i<100; ++i) {
				printf("%d ", readtimes[i]);
			}
			printf(" |%d\n", totalcount);
		}
	}
}

#include <thread>
using namespace std;
bool test_pipe() {
	int fds[2];

	pipe(fds);
//	if (evutil_socketpair(AF_UNIX, SOCK_STREAM, 0, fds) < 0) {
//		perror("socketpair");
//		abort();
//	}

	thread thr(consumer, fds[0]);

	producer(fds[1]);

	thr.join();
	return true;
}
